package com.peng.rpc.serializer;

import com.alibaba.nacos.shaded.io.grpc.protobuf.ProtoUtils;
import com.peng.rpc.enumeration.RpcError;
import com.peng.rpc.enumeration.SerializerCode;
import com.peng.rpc.exception.RpcException;
import io.protostuff.LinkedBuffer;
import io.protostuff.ProtostuffIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Wonderson
 * @email wpp.ds@foxmail.com
 * @date 2022-03-03 19:29
 */
@Slf4j
public class ProtobufSerializer implements CommonSerializer{

    private final LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
    private final Map<Class<?>, Schema<?>> schemaCache = new ConcurrentHashMap<>();

    @Override
    public byte[] serialize(Object object) {
        Class<?> aClass = object.getClass();
        Schema schema = getSchema(aClass);
        try {
            return ProtostuffIOUtil.toByteArray(object, schema, buffer);
        } catch (Exception e) {
            log.error("protobuf序列化失败：{}", object);
            throw new RpcException(RpcError.SERIALIZE_FAILED);
        } finally {
            buffer.clear();
        }
    }

    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        Schema schema = getSchema(clazz);
        Object res = schema.newMessage();
        ProtostuffIOUtil.mergeFrom(bytes, res, schema);
        return res;
    }

    @Override
    public int getCode() {
        return SerializerCode.PROTOBUF.getCode();
    }

    /**
     * @param clazz 序列化的目标类
     * @return 懒加载式的获取Schema
     */
    private Schema<?> getSchema(Class<?> clazz) {
        Schema<?> schema = schemaCache.get(clazz);
        if (Objects.isNull(schema)) {
            schema = RuntimeSchema.getSchema(clazz);
            if (Objects.nonNull(schema)) {
                schemaCache.put(clazz, schema);
            }
        }
        return schema;
    }
}
